package kafka
import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

import scala.collection.mutable.ArrayBuffer
object KafkaFa {


  /*
  数据格式:
      timestamp   area    city    userid  adid
      某个时间点 某个地区  某个城市   某个用户  某个广告

   */

  def createKafkaProducer: KafkaProducer[String, String] = {
    val props = new Properties
    // Kafka服务端的主机名和端口号
    props.put("bootstrap.servers", "had101:9092,had102:9092,had103:9092")
    // 等待所有副本节点的应答
    props.put("acks", "1")
    // 重试最大次数
    props.put("retries", "0")

    // 批消息处理大小
    props.put("batch.size", "16384")
    // 请求延时
    props.put("linger.ms", "1")
    //// 发送缓存区内存大小
    props.put("buffer.memory", "33554432")
    // key序列化
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    // value序列化
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    new KafkaProducer[String, String](props)
  }

  def main(args: Array[String]): Unit = {
    val topic = "tc"
    val producer: KafkaProducer[String, String] = createKafkaProducer
    while (true) {
        var msg="123"
          producer.send(new ProducerRecord(topic, msg))
          Thread.sleep(100)
        }
      }
      Thread.sleep(1000)

}
